package com.hdu.parallelUtils;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;


@Slf4j
public class ParallelExecuteTemplate {


    private static final Executor EXECUTOR = Executors.newFixedThreadPool(10);


    @SuppressWarnings("all")
    public static <K, V, Q> void parallelConsume(List<Q> list, int limitSize, ParallelConsume<Q> function) {

        if (list.size() <= limitSize) {
            function.execute(list);
        }

        int total = list.size();
        int threadNum = (total + limitSize - 1) / limitSize;
        CountDownLatch latch = new CountDownLatch(threadNum);
        Map<K, V> result = new ConcurrentHashMap<>();

        int start = 0;
        int end = 0;
        long curTime = System.currentTimeMillis();
        while (total > 0) {
            end = end + limitSize;
            asyncConsume(list, start, end, function, latch);
            start = end;
            total = total - limitSize;
        }
        try {
            latch.await();
            log.info("paramConsume success, cost : {} ms", System.currentTimeMillis() - curTime);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }


    @SuppressWarnings("all")
    public static <K, V, Q> Map<K, V> parallelGetMap(List<Q> list, int limitSize, ParallelGetMap<K, V, Q> function) {

        if (list.size() <= limitSize) {
            return new ConcurrentHashMap<>(function.execute(list));
        }

        int total = list.size();
        int threadNum = (total + limitSize - 1) / limitSize;
        CountDownLatch latch = new CountDownLatch(threadNum);
        Map<K, V> result = new ConcurrentHashMap<>();

        int start = 0;
        int end = 0;
        long curTime = System.currentTimeMillis();
        while (total > 0) {
            end = end + limitSize;
            asyncGetMap(list, start, end, function, result, latch);
            start = end;
            total = total - limitSize;
        }
        try {
            latch.await();
            log.info("parallelGetMap success, cost : {} ms", System.currentTimeMillis() - curTime);
            return result;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @SuppressWarnings("all")
    public static <V, Q> List<V> parallelGetList(List<Q> list, int limitSize, ParallelGetList<V, Q> function) {

        if (list.size() <= limitSize) {
            return function.execute(list);
        }

        int total = list.size();
        int threadNum = (total + limitSize - 1) / limitSize;
        CountDownLatch latch = new CountDownLatch(threadNum);
        List<V> finalResult = new ArrayList<>();

        int start = 0;
        int end = 0;
        long curTime = System.currentTimeMillis();
        while (total > 0) {
            end = end + limitSize;
            asyncGetList(list, start, end, function, finalResult, latch);
            start = end;
            total = total - limitSize;
        }
        try {
            latch.await();
            log.info("parallelGetList success, cost : {} ms", System.currentTimeMillis() - curTime);
            return finalResult;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }


    private static <Q> void asyncConsume(List<Q> list,
                                         int start,
                                         int end,
                                         ParallelConsume<Q> function,
                                         CountDownLatch latch) {
        long curTime = System.currentTimeMillis();
        final int realEnd = Math.min(end, list.size());
        EXECUTOR.execute(() -> {
            try {
                List<Q> curConsumeList = list.subList(start, realEnd);
                function.execute(curConsumeList);
                log.info("asyncConsume success, consume size : {}, cost : {} ms",
                        curConsumeList.size(),
                        System.currentTimeMillis() - curTime
                );
            } catch (Exception ignored) {

            } finally {
                latch.countDown();
            }
        });
    }

    private static <Q, K, V> void asyncGetMap(List<Q> list,
                                             int start,
                                             int end,
                                             ParallelGetMap<K, V, Q> function,
                                             Map<K, V> totalResult,
                                             CountDownLatch latch) {
        long curTime = System.currentTimeMillis();
        final int realEnd = Math.min(end, list.size());
        EXECUTOR.execute(() -> {
            try {
                List<Q> curQueryKeys = list.subList(start, realEnd);
                Map<K, V> curResult = function.execute(curQueryKeys);
                log.info("asyncGetMap success, query size : {}, cost : {} ms",
                        curQueryKeys.size(),
                        System.currentTimeMillis() - curTime
                );
                totalResult.putAll(curResult);
            } catch (Exception ignored) {

            } finally {
                latch.countDown();
            }
        });
    }

    private static <Q, V> void asyncGetList(List<Q> list,
                                          int start,
                                          int end,
                                          ParallelGetList<V, Q> function,
                                          List<V> totalResult,
                                          CountDownLatch latch) {
        long curTime = System.currentTimeMillis();
        final int realEnd = Math.min(end, list.size());
        EXECUTOR.execute(() -> {
            try {
                List<Q> curQueryKeys = list.subList(start, realEnd);
                List<V> curResult = function.execute(curQueryKeys);
                log.info("asyncGetList success, query size : {}, cost : {} ms",
                        curQueryKeys.size(),
                        System.currentTimeMillis() - curTime
                );
                totalResult.addAll(curResult);
            } catch (Exception ignored) {

            } finally {
                latch.countDown();
            }
        });
    }

    @FunctionalInterface
    public interface ParallelConsume<Q> {
        void execute(List<Q> list);
    }

    @FunctionalInterface
    public interface ParallelGetMap<K, V, Q> {
        Map<K, V> execute(List<Q> list);
    }

    @FunctionalInterface
    public interface ParallelGetList<V, Q> {
        List<V> execute(List<Q> list);
    }
}


